package com.kudu.sink;

import com.sun.javafx.collections.MappingChange;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kudu.client.*;

import java.util.Map;

public class MySinkToKudu extends RichSinkFunction<Map<String,Object>> {
    private String masterAddr;
    private String tableName;

    private KuduClient client;
    private KuduTable table;
    private KuduSession session;

    public MySinkToKudu(String masterAddr, String tableName) {
        this.masterAddr = masterAddr;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new KuduClient.KuduClientBuilder(masterAddr).build();
        table = client.openTable(tableName);
        session = client.newSession();
        session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
    }

    //schema: id String   name String   age int
    public void invoke(Map<String, Object> value, Context context) throws Exception {
        Insert insert = table.newInsert();
        PartialRow row = insert.getRow();
        row.addString("id",value.get("id").toString());
        row.addString("name",value.get("name").toString());
        row.addInt("age",Integer.valueOf(value.get("age").toString()));
        session.flush();
        session.apply(insert);
        session.close();
    }

    @Override
    public void close() throws Exception {
        client.close();
    }
}
